package com.atguigu.gmall.realtime.test;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author Felix
 * @date 2024/7/27
 * 该案例演示了FlinkSQL的join
 *                          左表                   右表
 *      内连接         onCreateAndWrite       onCreateAndWrite
 *      左外连接       onReadAndWrite         onCreateAndWrite
 *      右外连接       onCreateAndWrite       onReadAndWrite
 *      全外连接
 */
public class Flink01_SQL_Join {
    public static void main(String[] args) {
        //TODO 1.基本环境的准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        //TODO 2.从指定的网络端口读取员工数据数据并转换为动态表
        SingleOutputStreamOperator<Emp> empDS = env
                .socketTextStream("hadoop102", 8888)
                .map(
                        new MapFunction<String, Emp>() {
                            @Override
                            public Emp map(String lineStr) throws Exception {
                                String[] fieldArr = lineStr.split(",");
                                return new Emp(Integer.valueOf(fieldArr[0]), fieldArr[1], Integer.valueOf(fieldArr[2]), Long.valueOf(fieldArr[3]));
                            }
                        }
                );
        tableEnv.createTemporaryView("emp",empDS);
        //TODO 3.从指定的网络端口读取部门数据数据并转换为动态表
        SingleOutputStreamOperator<Dept> deptDS = env
                .socketTextStream("hadoop102", 8889)
                .map(
                        new MapFunction<String, Dept>() {
                            @Override
                            public Dept map(String lineStr) throws Exception {
                                String[] fieldArr = lineStr.split(",");
                                return new Dept(Integer.valueOf(fieldArr[0]), fieldArr[1], Long.valueOf(fieldArr[2]));
                            }
                        }
                );
        tableEnv.createTemporaryView("dept",deptDS);
        //TODO 4.内连接
        //注意：如果使用普通的内外连接，在底层会为参与连接的两张表各自维护一个状态，默认情况下，状态永不会失效
        //在生产环境下，一定一定一定要设置状态的失效时间  tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10))
        //tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e join dept d on e.deptno = d.deptno").print();

        //TODO 5.左外连接
        //如果是左外连接，左表数据先到，右表数据后到，会产生3条数据
        //左表    null    标记为+I
        //左表    null    标记为-D
        //左表    右表     标记为+I
        //tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e left join dept d on e.deptno = d.deptno").print();

        //TODO 6.右外连接
        //tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e right join dept d on e.deptno = d.deptno").print();

        //TODO 7.全外连接
        //tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e full join dept d on e.deptno = d.deptno").print();

        //TODO 8.将左外连接的结果写到Kafka主题中
        //必须使用upsert-kafka连接器
        //如果左表数据先到，右表数据后到，kafka主题会接收到3条消息
        // 左表   null
        // null
        // 左表   右表
        //如果使用FlinkSQL从kafka主题中读取数据的话，会自动过滤空消息；如果使用的FlinkAPI从kafka主题中读取数据的话，默认的SimpleStringSchema
        //是处理不了空消息的，需要自定义反序列化器。
        //其中第一条数据和第三条数据，属于重复数据，后续（DWS）需要对重复数据进行去重操作
        tableEnv.executeSql("CREATE TABLE emp_dept (\n" +
                "  empno integer,\n" +
                "  ename string,\n" +
                "  deptno integer,\n" +
                "  dname string,\n" +
                "  PRIMARY KEY (empno) NOT ENFORCED\n" +
                ")WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = 'first',\n" +
                "  'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")");
        tableEnv.executeSql("insert into emp_dept select e.empno,e.ename,d.deptno,d.dname from emp e left join dept d on e.deptno = d.deptno");


    }
}
